package org.pupil.rpc.registry.registryImpl;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.pupil.rpc.common.RpcServiceNameBuilder;
import org.pupil.rpc.common.ServiceMeta;
import org.pupil.rpc.config.RpcProperties;
import org.pupil.rpc.registry.RegistryService;

import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

/**
 * zk注册中心
 */
public class ZookeeperRegistry implements RegistryService {

    // 连接失败等待重试时间
    public static final int BASE_SLEEP_TIME_MS = 1000;
    // 重试次数
    public static final int MAX_RETRIES = 3;
    // 服务注册根路径
    public static final String ZK_BASE_PATH = "/pupil_rpc";
    // 用于注册、发现和管理服务
    private final ServiceDiscovery<ServiceMeta> serviceDiscovery;

    /**
     * 启动zk
     * @throws Exception
     */
    public ZookeeperRegistry() throws Exception{
        // 注册中心地址
        String registerAddr = RpcProperties.getInstance().getRegisterAddr();
        // 使用Curator 框架创建一个 ZooKeeper 客户端对象，设置地址 和 重试策略（初始睡眠时间，最大重试次数）
        CuratorFramework client = CuratorFrameworkFactory.newClient(registerAddr, new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRIES));
        client.start(); // 启动客户端
        // 序列化服务实例的序列化器(Json 格式)
        JsonInstanceSerializer<ServiceMeta> serializer = new JsonInstanceSerializer<>(ServiceMeta.class);
        // 创建一个服务发现的对象
        this.serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMeta.class) // 对象构造器
                .client(client) // 设置使用的 ZooKeeper 客户端
                .serializer(serializer) // 设置用于序列化服务实例的序列化器
                .basePath(ZK_BASE_PATH) // 设置服务注册的根路径
                .build();
        this.serviceDiscovery.start();
    }


    /**
     * 服务注册
     * @param serviceMeta 服务数据
     * @throws Exception
     */
    @Override
    public void register(ServiceMeta serviceMeta) throws Exception {
        // 服务实例
        ServiceInstance<ServiceMeta> serviceInstance = ServiceInstance
                .<ServiceMeta>builder() // 对象构造器
                .name(RpcServiceNameBuilder.buildServiceKey(serviceMeta.getServiceName(), serviceMeta.getServiceVersion())) // 设置服务实例的名称
                .address(serviceMeta.getServiceAddr()) // 设置服务实例的地址
                .port(serviceMeta.getServicePort()) // 设置服务实例的端口
                .payload(serviceMeta) // 设置服务实例的负载，传入服务信息用于负载均衡
                .build();
        // 注册服务
        serviceDiscovery.registerService(serviceInstance);
    }

    /**
     * 获取所有服务
     * @param serviceName
     * @return
     * @throws Exception
     */
    private List<ServiceMeta> listServices(String serviceName) throws Exception{
        // 获取所有服务实例
        Collection<ServiceInstance<ServiceMeta>> serviceInstances = serviceDiscovery.queryForInstances(serviceName);
        // 获取所有服务节点
        List<ServiceMeta> serviceMetas = serviceInstances.stream().map(serviceMetaServiceInstance -> serviceMetaServiceInstance.getPayload()).collect(Collectors.toList());
        return serviceMetas;
    }

    /**
     * 服务注销
     * @param serviceMeta
     * @throws Exception
     */
    @Override
    public void unRegister(ServiceMeta serviceMeta) throws Exception {
        ServiceInstance<ServiceMeta> serviceInstance = ServiceInstance
                .<ServiceMeta>builder()
                .name(serviceMeta.getServiceName())
                .address(serviceMeta.getServiceAddr())
                .port(serviceMeta.getServicePort())
                .payload(serviceMeta)
                .build();
        serviceDiscovery.unregisterService(serviceInstance);
    }

    /**
     * 获取所有服务节点
     * @param serviceName
     * @return
     */
    @Override
    public List<ServiceMeta> discoveries(String serviceName) {
        try {
            return listServices(serviceName);
        } catch (Exception e) {
            e.printStackTrace();
        }
        // 发生错误时，返回空列表
        return Collections.EMPTY_LIST;
    }

    /**
     * 关闭
     * @throws IOException
     */
    @Override
    public void destroy() throws IOException {
        serviceDiscovery.close();
    }
}
